package hust.rpc.handler;

import hust.rpc.constants.Constants;
import hust.rpc.pojo.Message;
import hust.rpc.pojo.RpcRequest;
import hust.rpc.pojo.RpcResponse;
import hust.rpc.serialize.SerializeUtils;
import hust.rpc.serialize.Serializer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;

import java.util.Arrays;

/**
 * 4B MagicNumber(魔数) "hust"
 * 1B Version(RPC版本号)
 * 1B Codec(编码方式)
 * 1B MessageType(消息类型)
 * 4B FullLength(消息总长度) -> 32bit -> 2^32byte -> 4GB
 * ** Body(消息数据,Object类型)
 */
@Slf4j
public class MessageDecoder extends LengthFieldBasedFrameDecoder {

    public MessageDecoder() {
        // 最大长度: 8MB
        // 长度字段偏移: 7byte
        // 长度字段长度: 4byte
        // 长度字段调整至: 消息总长度 - 消息头长度
        // 不跳过任意byte
        this(Constants.MAX_FRAME_LENGTH, 7, 4, -Constants.HEADER_LENGTH, 0);
    }

    public MessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip) {
        super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip);
    }

    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        Object decoded = super.decode(ctx, in);
        if (decoded instanceof ByteBuf) {
            ByteBuf frame = (ByteBuf) decoded;
            if (frame.readableBytes() >= Constants.HEADER_LENGTH) {
                try {
                    return decodeFrame(frame);
                } catch (Exception e) {
                    log.error("消息编码失败: ", e);
                    throw e;
                } finally {
                    frame.release();
                }
            }
        }
        return decoded;
    }

    /**
     * 解码消息帧
     *
     * @param frame ByteBuf
     * @return Object
     */
    private Object decodeFrame(ByteBuf frame) {
        // 校验魔数 4byte
        checkMagicNumber(frame);
        // 校验RPC框架版本号 1byte
        byte version = checkVersion(frame);
        // 获取消息编码方式 1byte
        byte codec = frame.readByte();
        // 获取消息类型信息 1byte
        byte messageType = frame.readByte();
        // 获取消息长度信息 int = 4byte
        // 消息体长度 = 总长度 - 消息头长度
        int bodyLength = frame.readInt() - Constants.HEADER_LENGTH;

        // 构建统一封装消息对象
        Message message = Message.builder()
                .codec(codec)
                .version(version)
                .messageType(messageType)
                .build();

        // 消息体长度非空
        if (bodyLength > 0) {
            byte[] body = new byte[bodyLength];
            // 获取消息体字节数组
            frame.readBytes(body);
            // 根据编码方式获取解码器
            Serializer serializer = SerializeUtils.getSerializer(codec);
            // 根据消息类型进行解码
            if (messageType == Constants.REQUEST_TYPE) {
                RpcRequest request = serializer.deserialize(body, RpcRequest.class);
                message.setData(request);
            } else {
                RpcResponse response = serializer.deserialize(body, RpcResponse.class);
                message.setData(response);
            }
        }
        return message;
    }

    /**
     * 校验版本号
     *
     * @param frame ByteBuf
     */
    private byte checkVersion(ByteBuf frame) {
        byte version = frame.readByte();
        if (version > Constants.VERSION) {
            throw new IllegalArgumentException("版本号不正确: " + version);
        }
        return version;
    }

    /**
     * 校验魔数
     *
     * @param frame ByteBuf
     */
    private void checkMagicNumber(ByteBuf frame) {
        int magicNumberLength = Constants.MAGIC_NUMBER.length;
        byte[] magicNumber = new byte[magicNumberLength];
        frame.readBytes(magicNumber);
        for (int i = 0; i < magicNumberLength; i++) {
            if (magicNumber[i] != Constants.MAGIC_NUMBER[i]) {
                throw new IllegalArgumentException("魔数不正确: " + Arrays.toString(magicNumber));
            }
        }
    }
}